<?php
namespace Handler;

use Core\TanbaSwooleTable;
use Core\Model;
use Core\TbRedis;
use Helper\kfRssi;
use Helper\GetLocationPoint;
use Mockery\Exception;

/**
 * Class BaseDataServer
 * @package Core
 * @author shiquanWu
 * BaseDataServer进程池，目前主要实现以下数据服务能力：
 * 批量写入设备信号数据到数据表
 *
 *
 */
class BaseDataServer
{
    const QUEUE_LOCK_TIMEOUT = 60;

    const WORKER_PROCESS_SLEEP_SECOND = 1;

    public $mpid = 0;

    protected $KfRssi = array();

    // Counter
    protected $a000_atomic;
    protected $a001_atomic;
    protected $a002_atomic;

    public $a000_data_map = [];
    public $a001_data_map = [];
    public $a002_data_map = [];

    private $getPointLoction;

    /**
     * 初始化构造函数
     * 1. 初始化系统配置
     * 2. 初始计数器用来锁定资源
     */
    public function __construct()
    {
        try {
            cli_set_process_title("basedataserver_master_process");
            $this->mpid = posix_getpid();
            echo "Duzun BaseDataServer master process is started on: " . date('Y-m-d H:i:s') . "!\n";
            $logFile = dirname(dirname(__FILE__)) . '/log/basedata.pid';
            file_put_contents($logFile, $this->mpid);

            // init counter
            $this->a000_atomic = new \Swoole\atomic(0);
            $this->a001_atomic = new \Swoole\atomic(0);
            $this->a002_atomic = new \Swoole\atomic(0);

//            if(config('openLocation'))
//                $this->getPointLoction = new GetLocationPoint();

        } catch(\Exception $e) {
            die('Duzun BaseDataServer process pool fails to start:' . $e->getMessage());
        }
    }

    /**
     * 工作进程启动回调函数
     * @param  Swoole\Process\Pool $pool Pool对象
     * @param  int $workerId 当前工作进程的编号
     */
    public function onWorkerStart($pool, $workerId)
    {
        cli_set_process_title("basedataserver_worker_process#{$workerId}");
        echo "Duzun BaseDataServer worker process#{$workerId} is started!" . PHP_EOL;

        while (true) {
            // Connects redis server
            while (true) {
                try {
                    if (false == $theWorkerRedis = @TbRedis::init('base_data_server_proc_' . $workerId)) {
                        echo "Redis base_data_proc_{$workerId} connects faild!" . PHP_EOL;
                        throw new \Exception("Redis connects faild");
                    }
                    echo "Redis base_data_proc_{$workerId} connects ok" . PHP_EOL;
                    break;
                } catch (\Exception $e) {
                    echo "Reconnect redis server …… " . PHP_EOL;
                    sleep(5);
                }
            }

            // Connects mysql server
            while (true) {
                try {
                    if (false == $conn = @Model::init('base_data_server_proc_' . $workerId)) {
                        echo "Db base_data_proc_{$workerId} connects faild!" . PHP_EOL;
                        throw new \Exception("DB connects faild");
                    }
                    echo "Db base_data_proc_{$workerId} connects ok" . PHP_EOL;
                    break;
                } catch (\Exception $e) {
                    echo "Reconnect Db server …… " . PHP_EOL;
                    sleep(5);
                }
            }

            while (true) {
                try {
                    foreach (TYPE_LISTS as $typeKey) {
                        $objAtomic = $typeKey . '_atomic';
                        $typeRedisQueueKey = getRedisTypeKey($typeKey);
                        $listQueueLength = intval($theWorkerRedis->lLen($typeRedisQueueKey));
                        if ($listQueueLength > 0 && time() - $this->$objAtomic->get() >= config('pool')[$typeKey . '_queue_consume_timeout']) {
                            $this->$objAtomic->set(time()); // Lock
                            $dataMapKey = $typeKey . '_data_map';

                            // The worker process continuously fetches data from the specified type of data queue until the buffer container has saved all the data for some timestamp
                            while ($queueVal = $theWorkerRedis->rPop($typeRedisQueueKey)) {
                                $dataArr = json_decode($queueVal, true);
                                if (!in_array($dataArr['type'], TYPE_LISTS)) continue;
                                if (count($this->$dataMapKey) >= config('pool')[$typeKey . '_queue_total_size'] && !isset($this->$dataMapKey[$dataArr['time']])) {
                                    $theWorkerRedis->rPush($typeRedisQueueKey, $queueVal);
                                    $consumeDataArr = $this->filterDataMapByType($dataArr['type']);
                                    break;
                                } else {
                                    if (isset($this->$dataMapKey[$dataArr['time']])) {
                                        $this->$dataMapKey[$dataArr['time']][$dataArr['mac'] . '-' . $dataArr['rssi'] . '-' . $dataArr['time']] = $dataArr;
                                    } else {
                                        $this->$dataMapKey[$dataArr['time']][$dataArr['mac'] . '-' . $dataArr['rssi'] . '-' . $dataArr['time']] = $dataArr;
                                    }
                                }
                            }
                            $this->$objAtomic->set(0); // Forced timeout

                            // Work process reconfiguration data
                            if (isset($consumeDataArr) && !empty($consumeDataArr)) {
                                $this->filterConsumeData($typeKey, $consumeDataArr, $theWorkerRedis, $conn);
                                unset($consumeDataArr);// 清除数据，防止复用
                            }
                            echo "------Task worker(" . $workerId . ") finish task ------" . PHP_EOL . PHP_EOL;
                        }
                        sleep(self::WORKER_PROCESS_SLEEP_SECOND);
                    }
                } catch (\Exception $e) {
                    $errMsg = "---Task worker(" . $workerId . ") fail to run: " . $e->getMessage() . PHP_EOL;
                    echo $errMsg;
                    TbRedis::close('base_data_server_proc_' . $workerId);
                    break;

                }
            }
            sleep(5);
        }
    }

    /**
     * 工作进程结束回调函数
     * @param  Swoole\Process\Pool $pool Pool对象
     * @param  int $workerId 当前工作进程的编号
     */
    public function onWorkerStop($pool, $workerId)
    {
        $quitMsg = "basedataserver_worker_process#{$workerId} is stopped\n";
        echoMsg($quitMsg, null, true, '', 'base_data_server_' . date('Y_m_'));
    }

    /**
     * 对待写入数组重组排列
     * @param string $typeKey 设备数据类型
     * @param array $consumeDataArr 待写入数组
     * @param object $theWorkerRedis Redis链接
     * @param object $conn Db连接
     * @return void
     */
    public function filterConsumeData($typeKey, $consumeDataArr, $theWorkerRedis, $conn)
    {
        $insertDataArr = array();
        foreach ($consumeDataArr as $timeData) {
            foreach ($timeData as $data) {
                // Check if exists tbid
                $data['tbid'] = isset($data['tbid']) ? $data['tbid'] : $theWorkerRedis->hGet($data['tbmac'], 'id');
                if (isset($data['tbid']) && !empty($data['tbid'])) {
                    $dbData = $data;
                    // Filter columns
                    if ($data['type'] == TYPE_STA && isset($data['ap_brand'])) {
                        unset($dbData['dvuid'], $dbData['apuid'], $dbData['ap_brand'], $dbData['dv_brand']);
                    }
                    unset($dbData['tbmac'], $dbData['type'], $dbData['devkey'], $dbData['find'], $dbData['devid'], $dbData['json']);
                    $dbData['sfind'] = $data['sfind'] = isset($data['sfind']) && $data['sfind'] != 0 ? $data['sfind'] : 1;
                    $insertDataArr[] = $dbData;
                    //$this->setBaseInfo($data, $theWorkerRedis, $conn); // Base service
                }
            }
        }

        // Inserts data into three data tables(a001, a002, a003)
        $this->dealIntoDb($typeKey, $theWorkerRedis, $insertDataArr, $conn);
    }


    /**
     * 批量写入AP数据表、STA数据表、RF数据表
     * @param string $type 设备类型
     * @param object $redisConn redis连接对象
     * @param object $dbConn db连接对象
     */
    public function dealIntoDb($type, $redisConn, $insertData, $dbConn)
    {
        if (in_array($type, TYPE_LISTS)) {
            // Write to data tables
            $result = !empty($insertData) ? $dbConn->insert($type, $insertData) : false;
            if (!$result) {
                $errMsg = "\n DB table `" . $type . "` has an insert error: " . json_encode($dbConn->error()). "\n";
                echo $errMsg;
                echoMsg($errMsg, null, true, '', 'base_data_server_' . date('Y_m_'));
            } else {
                $msg = "\n # DB table `" . $type . "` inserts OK # \n";
                echo $msg;
                //echoMsg($msg, null, true, '', 'base_data_server_' . date('Y_m_'));
            }
        }
    }

    /**
     * 从数据缓冲容器中抽取待写入数据集合
     * @param string $dataType 设备数据类型
     * @return array $getDataArr 待写入数据数组
     */
    private function filterDataMapByType($dataType)
    {
        $dataMapKey = $dataType . '_data_map';
        $timeDataArr = $this->$dataMapKey;
        ksort($timeDataArr);
        $getDataArr = array_slice($timeDataArr, 0, config('pool')[$dataType . '_queue_consume_size'], true);
        $reserveDataArr = array_diff_key($timeDataArr, $getDataArr);
        $this->$dataMapKey = $reserveDataArr;
        return $getDataArr;
    }

    /**
     * 每次接收的数据拆分入组，批量写入AP指定时间探测表、STA指定时间探测表、RF指定时间探测表
     * @param array $data 接收的数据
     * @param object $theRedis Redis链接
     * @param object $theDb Db连接
     * @return array
     * Created by: shiquanWu
     */
    private function setBaseInfo($data, $theRedis, $theDb)
    {
        if (empty($data)) return false;
        $devkey     = $data['devkey'];
        $tbdeviceID = $data['tbmac'];
        $tbid       = $data['tbid'];
        $data['rssi'] = abs($data['rssi']);
        $sendssi       = $data['type'] == TYPE_RF ? abs($data['sendRssi']) : $data['rssi'];
        $data['sendRssi'] = $sendssi;

        // Reduce redis initialization or update state
        $rdskey = DzDataType($data['type']) . "_" . $devkey;
        $dvid   = setNewDeviceInfo($data, $theRedis, $theDb);
        if (!isset($dvid) || empty($dvid)) return;
        $theRedis->hMset($tbdeviceID, array('updatetime' => time()));
        $data['devid'] = $dvid;

        if (config('openKF')) {
            if (!isset($this->KfRssi[$tbid . "_" . $devkey]))
                $this->KfRssi[$tbid . "_" . $devkey] = new kfRssi(); // notice deletion
            $data['realRssi'] = floatval(sprintf("%.2f", $this->KfRssi[$tbid . "_" . $devkey]->kfRssi($sendssi)));

            $tbinfo = gettbinfo($tbid, $theRedis);
            if (empty($tbinfo['rn'])) $tbinfo['rn'] = 1;
            if (empty($tbinfo['a0'])) $tbinfo['a0'] = 45;
            $distdata = array(
                'realRssi' => $data['realRssi'],
                'rssi_n' => $tbinfo['rn'],
                'rssi_a' => $tbinfo['a0'],
            );
            $data['dist'] = floatval(getDistancefromSdssi($distdata));
        } else {
            $data['realRssi'] = $data['sendRssi'];
            $data['dist'] = 0;
        }

        // Calculate the total number and signal of terminal detected by different tanba in 60 seconds.
        $devitem = $theRedis->hGetAll($rdskey);
        if (isset($devitem['staytime']) && !empty($devitem['staytime']) && $devitem['staytime'] > 1) {
            $stayTime = $devitem['staytime'];
            if ($data['time'] - $stayTime >= 60 && isset($devitem['stayinfo']) && !empty($devitem['stayinfo'])) {
                $stanct = json_decode($devitem['stayinfo'], true);
                if (isset($stanct) && count($stanct) > 0) {
                    foreach ($stanct as $k => $v) {
                        $insertData[] = [
                            'tbid' => $k,
                            'deviceid' => $dvid,
                            'findcount'=> (int)$v['cnt'],
                            'rssi' => json_encode($v['rssi']),
                            'kmrssi' => json_encode($v['kmRssi']),
                            'kmdist' => json_encode($v['kmdist']),
                            'createtime' => $stayTime,
                        ];
                    }

                    // Write to short tables
                    $insertResult = $theDb->insert('short_' . $data['type'], $insertData);
                    $theRedis->hMset($rdskey, array('staytime' => $data['time'], 'stayinfo' => ''));
                    if (!$insertResult) {
                        $errMsg = "\n DB table `short_" . $data['type'] . "` has an insert error: " . $theDb->error() . "\n";
                        echo $errMsg;
                        echoMsg($errMsg, null, true, '', 'base_data_server_' . date('Y_m_'));
                    } else {
                        $msg = "\n # DB table `short_" . $data['type'] . "` inserts OK # \n";
                        echo $msg;
                    }
                }
            } else {
                if (isset($devitem['stayinfo'])) {
                    if (empty($devitem['stayinfo'])) $devitem['stayinfo'] = json_encode([]);
                    $stanct = json_decode($devitem['stayinfo'], true);
                    if (isset($stanct[$tbid]) && !empty($stanct[$tbid]['cnt'])) {
                        $stanct[$tbid]['cnt'] = (int)$stanct[$tbid]['cnt'] + $data['sfind'];
                    } else {
                        $stanct[$tbid]['cnt'] = $data['sfind'];
                    }
                } else {
                    $stanct[$tbid]['cnt'] = $data['sfind'];
                }

                $stanct[$tbid]['id'] = $tbid;
                $stanct[$tbid]['rssi'][] = $sendssi;
                $stanct[$tbid]['kmRssi'][] = $data['realRssi'];
                $stanct[$tbid]['kmdist'][] = $data['dist'];
                $theRedis->hSet($rdskey, 'stayinfo', json_encode($stanct));
            }
        } else {
            $stanct[$tbid]['cnt']= $data['sfind'];
            $stanct[$tbid]['id'] = $tbid;
            $stanct[$tbid]['rssi'][] = $sendssi;
            $stanct[$tbid]['kmRssi'][] = $data['realRssi'];
            $stanct[$tbid]['kmdist'][] = $data['dist'];
            $theRedis->hMset($rdskey, array('staytime'=> $data['time'], 'stayinfo' => json_encode($stanct)));
        }
        echo ".";
    }
}
